package Top4.transaction;

import com.rabbitmq.client.*;
import util.ConnectionUtil;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

/**
 * @Auther: Administrator
 * @Date: 2019/2/28 20:09
 * @Description: 死信交换机  生产者首先发送一条携带路由键为 " transaction " 的消息，然后经过交换器
 * exchange .normal 顺利地存储到队列 queue.normal 中 。由于队列 queue.normal 设置了过期时间为
 * 10s ， 在这 10s 内没有消费者消费这条消息，那么判定这条消息为过期。由于设置了 transaction ， 过期
 * 之时 ， 消息被丢给交换器 exchange.transaction 中，这时找到与 exchange.transaction 匹配的队列 queue .transaction ， 最
 * 后消息被存储在 queue.transaction 这个死信队列中。
 */
public class TransactionProducer {
    private static final String EXCHANGE_NAME = "transaction_exchange";
    private static final String ROUTING_KEY = "routingkey_demo";
    private static final String QUEUE_NAME = "transaction_queue";
    private static final String IP_ADDRESS = "192.168.204.131";
    private static final int PORT = 5672;//RabbitMQ 服务端默认端口号为 5672

    public static void main(String[] args) throws IOException, TimeoutException {
    /*    ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost(IP_ADDRESS);
        connectionFactory.setPort(PORT);
        connectionFactory.setUsername("root");
        connectionFactory.setPassword("root");
             //创建连接
        Connection connection = connectionFactory.newConnection();*/

        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();

        //定义普通的交换机
        channel.exchangeDeclare("exchange.normal", "fanout", true);
        //定义死信交换机
        channel.exchangeDeclare("exchange.transaction", "direct", true);
        Map<String, Object> map = new HashMap<String, Object>();
        map.put("x-message-ttl", 10000);
        map.put("x-dead-letter-exchange", "exchange.transaction");
        map.put("x-dead-letter-routing-key", "routingKey");
        //普通的队列并绑定
        channel.queueDeclare("queue.normal", true, false, false, map);
        channel.queueBind("queue.normal", "exchange.normal", "");
        //死信队列
        channel.queueDeclare("queue.transaction", true, false, false, null);
        channel.queueBind("queue.transaction", "exchange.transaction", "routingKey");
        try {
            //        开启事务
            channel.txSelect();
            channel.basicPublish("exchange.normal", "rk", MessageProperties.PERSISTENT_TEXT_PLAIN, "transaction message".getBytes());
            System.out.println("消息发送成功");
            /* 提交事务*/
            int result = 1 / 0;
            channel.txCommit();
        } catch (ArithmeticException e) {
            e.printStackTrace();
            channel.txRollback();
        }
        channel.close();
        connection.close();
    }
}
